package com.pluser.mq.engine;

import java.io.IOException;
import java.util.Map;

import com.ibm.mq.MQException;
import com.ibm.mq.MQGetMessageOptions;
import com.ibm.mq.MQMessage;
import com.ibm.mq.constants.MQConstants;
import com.pluser.mq.vo.QueueInfo;
import com.pluser.mq.vo.QueueManagerInfo;

/**
 * 
 * ConditionReceiver class
 * @author pluser
 * @version 1.0
 * @date 2014-05-03
 * @modifiedHis
 *
 * @description
 * listening receiver queue, get message from queue with
 * message comparator
 */
public class ConditionReceiver {

	public static final int MQOO_REMOVE_ON_BROWSE = MQConstants.MQOO_INPUT_SHARED
			| MQConstants.MQOO_BROWSE | MQConstants.MQOO_FAIL_IF_QUIESCING;
	public static final int MQOO_BROWSE = MQConstants.MQOO_BROWSE
			| MQConstants.MQOO_FAIL_IF_QUIESCING;
	public static final int MQGMO_REMOVE_ON_BROWSE = MQConstants.MQGMO_MSG_UNDER_CURSOR
			| MQConstants.MQGMO_FAIL_IF_QUIESCING;
	public static final int MQGMO_BROWSE = MQConstants.MQGMO_BROWSE_NEXT
			| MQConstants.MQGMO_WAIT | MQConstants.MQGMO_FAIL_IF_QUIESCING;
	
	private MQGetMessageOptions gmo;
	private IMessageComparator msgComparator;
	private Receiver receiver;
	private boolean isRemove;
	
	/**
	 * constructor
	 * @param qmInfo
	 * @param qName qName name of queue
	 * @param msgComparator
	 * @param isRemove
	 * @throws MQException
	 */
	public ConditionReceiver(QueueManagerInfo qmInfo, String qName,
			IMessageComparator msgComparator, boolean isRemove)
			throws MQException {
		
		this.receiver = new Receiver(qmInfo, getQueueInfo(qName, isRemove));		
		this.msgComparator = msgComparator;
		this.isRemove = isRemove;
		
		gmo = new MQGetMessageOptions();
		gmo.options = MQGMO_BROWSE;
	}
	
	/**
	 * get queue info
	 * @param qName name of queue
	 * @param isRemove get message whether remove this message from queue
	 * @return
	 */
	private QueueInfo getQueueInfo(String qName, boolean isRemove) {
		
		QueueInfo qInfo = new QueueInfo(qName);
		
		if(isRemove) {
			qInfo.setOpenOptions(MQOO_REMOVE_ON_BROWSE);
		} else {
			qInfo.setOpenOptions(MQOO_BROWSE);
		}
		
		return qInfo;
	}
	
	/**
	 * remove message on browse queue
	 * @return
	 * @throws MQException
	 */
	private MQMessage removeMsgOnBrowse() throws MQException {
		
		MQGetMessageOptions gmo = new MQGetMessageOptions();
		MQMessage msg = new MQMessage();
		
		gmo.options = MQGMO_REMOVE_ON_BROWSE;
		receiver.getMsg(msg, gmo);
		
		return msg;
	}
	
	/**
	 * get message from queue
	 * @param orgCmpMap original compare data container
	 * @param waitInterval wait time, millisecond
	 * @return
	 * @throws IOException
	 * @throws MQException
	 */
	public byte[] getMsg(Map<String, String> orgCmpMap, int waitInterval)
			throws IOException, MQException {
		
		byte[] msgBytes = null;
		
		gmo.waitInterval = waitInterval;
		
		while(true) {
			
			msgBytes = receiver.getMsg(gmo);
			
			if(msgComparator.compare(msgBytes, orgCmpMap)) {
				
				if(isRemove) {
					removeMsgOnBrowse();
				}
				
				break;
			}
		}
		
		return msgBytes;
	}
	
	/**
	 * disconnect, close queue and queue manager
	 * @throws MQException
	 */
	public void disconnect() throws MQException {
		
		if(receiver != null) {
			receiver.disconnect();
		}
	}
	
}
